package com.lry.router;

import com.alibaba.fastjson.JSONObject;
import com.lry.*;
import lombok.Data;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Data
public class Subscribe {

    private Map<String, List<BrokerInfo>> brokerMap = new ConcurrentHashMap<>();

    private String registerType;

    private String registerAddr;

    private BrokerInfoChangeListener brokerInfoChangeListener;

    public Subscribe(String registerType, String registerAddr,BrokerInfoChangeListener brokerInfoChangeListener) {
        this.registerType = registerType;
        this.registerAddr = registerAddr;
        this.brokerInfoChangeListener = brokerInfoChangeListener;
    }

    public void subscribe(){

        RegistryService registry = RegistryFactory.getRegistry(registerType, registerAddr);

        List<String> contents = registry.subscribe(content-> {

            BrokerInfo brokerInfo = JSONObject.parseObject(content, BrokerInfo.class);

            brokerInfoChangeListener.change(brokerInfo);
        });

        for (String content : contents) {
            BrokerInfo brokerInfo = JSONObject.parseObject(content, BrokerInfo.class);
            brokerMap.compute(brokerInfo.getBrokerName(),(k,v)->{
                if(null==v){
                    v = new ArrayList<>();
                }
                v.add(brokerInfo);
                return v;
            });
        }
    }


    public List<BrokerInfo> getMasters(){
        List<BrokerInfo> masters = new ArrayList<>();
        for (String brokerName : brokerMap.keySet()) {
            for (BrokerInfo brokerInfo : brokerMap.get(brokerName)) {
                if(brokerInfo.isMaster()){
                    masters.add(brokerInfo);
                }
            }
        }
        return masters;
    }

}
